/**
 * @file STL_Mqtt.cpp
 * 
 * @author Derek Liu (yjkhtddx@sina.com)
 * @brief 
 * @version 0.1
 * @date 2020-09-23
 * 
 * @copyright Copyright Sci-Tech Laboratories (c) 2020
 * 
 */

#include "STL_Mqtt.h"
#include <thread>
#include <mutex>
#include <functional>
#include "STLOG.h"
#include "STL_Text.h"
#include <mosquittopp/mosquittopp.h>
#include "STL_Async.h"
#include <string.h>

// Default SSL properties
#define TLS_VERSION NULL  //"tlsv1.2"
#define SSL_VERIFY_NONE 0 // the server will not be verified in any way.
#define SSL_VERIFY_PEER 1 // the server certificate will be verified
#define CIPHER NULL

namespace STL
{

    /**
     * Establish TLS connection with mosquitto MQTT instance using CA cert file
     * @param mosq mosquitto MQTT instance
     * @param cafile path to CA cert file
     * @return success establishing TLS connection
     */
    int set_tls_connection(struct mosquitto *mosq, const char *cafile, bool isVerifyPeer)
    {
        int retval;

        mosquitto_int_option(mosq, MOSQ_OPT_SSL_CTX_WITH_DEFAULTS, 1);

        retval = mosquitto_tls_set(mosq, cafile, NULL, NULL, NULL, NULL);
        if (retval == MOSQ_ERR_SUCCESS)
        {
            retval = mosquitto_tls_insecure_set(mosq, true);
            if (retval == MOSQ_ERR_SUCCESS)
            {
                int isVerify = SSL_VERIFY_PEER;
                if (!isVerifyPeer)
                    isVerify = SSL_VERIFY_NONE;
                retval = mosquitto_tls_opts_set(mosq, isVerify, TLS_VERSION, CIPHER);
                if (retval != MOSQ_ERR_SUCCESS)
                {
                    printf("mosquitto_tls_opts_set failed, %s\n", mosquitto_strerror(retval));
                }
            }
            else
            {
                printf("mosquitto_tls_insecure_set failed, %s\n", mosquitto_strerror(retval));
            }
        }
        else
        {
            printf("mosquitto_tls_set failed, %s\n", mosquitto_strerror(retval));
        }

        return retval;
    }

    int STL_Mqtt::s_SubscribeMessageIndex = 0;
    std::mutex STL_Mqtt::s_init_mutex;
    void STL_Mqtt::initlib()
    {
        std::lock_guard<std::mutex> guard(s_init_mutex);
        static bool Initialized = false;
        if (Initialized)
            return;

        // step1：初始化库,必须在调用库函数之前调用。 Must be called before any other mosquitto functions.
        mosquitto_lib_init();
        Initialized = true;
    }

    std::string STL_Mqtt::toLowerHex(std::string str, size_t offset /*= 0*/, size_t max /*= 0*/)
    {
        std::string ret = "";
        char tempBuffer[4];
        for (std::size_t i = offset; i < str.length(); i++)
        {
            if (max > 0 && i - offset >= max)
                break;
            uint8_t temp = str.at(i);
            sprintf(tempBuffer, "%02x", temp);
            ret.append(std::string(tempBuffer));
        }
        return ret;
    }

    STL_Mqtt::STL_Mqtt(STL_Mqtt::Option option)
        : m_option(option), m_status(STL_Mqtt::Status::None)
    {
        static int s_InstanceIndex = 0;
        m_index = s_InstanceIndex++;
    }

    int STL_Mqtt::start()
    {
        m_run = true;
        setStatus(STL_Mqtt::Status::Init);
        m_thread = std::thread(std::bind(&STL_Mqtt::mqttThread, this));
        return 0;
    }
    int STL_Mqtt::stop()
    {
        m_run = false;
        //TODO 断开连接,停掉线程,注销对象
        return 0;
    }

    void STL_Mqtt::setStatus(STL_Mqtt::Status status)
    {
        {
            std::lock_guard<std::mutex> guard(m_mutex);
            m_status = status;
        }
        if (m_statusCallback)
            m_statusCallback(m_status);
    }

    int STL_Mqtt::initSubscribe()
    {
        std::lock_guard<std::mutex> guard(m_subscribeMapMutex);
        for (auto ite = m_subscribeMap.begin(); ite != m_subscribeMap.end();)
        {
            auto it = ite++;
            auto topic = it->first;
            auto info = it->second;
            //第三部：订阅
            {
                std::lock_guard<std::mutex> guard(m_mutex);
                if (mosquitto_subscribe(m_pMqtt_st, NULL, topic.c_str(), 0) == MOSQ_ERR_SUCCESS)
                {
                    modulelog_debug(STL_Text::format("mqtt subscribe <topic:%s>[%s](%s:%d) succeed\n", topic.c_str(), m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort));
                }
                else
                {
                    modulelog_debug(STL_Text::format("mqtt subscribe <topic:%s>[%s](%s:%d) failed\n", topic.c_str(), m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort));
                }
            }
        }
    }

    int STL_Mqtt::subscribe(std::string topic, int qos, STL_MqttDelegate *pDelegate /* = nullptr*/)
    {
        //第一步：检测订阅
        std::lock_guard<std::mutex> guard(m_subscribeMapMutex);
        if (m_subscribeMap.count(topic) == 1) // 主题已订阅
        {
            if (m_subscribeMap[topic].count(pDelegate) == 1) // 委托对象已订阅
            {
                modulelog_warn(STL_Text::format("topic[%s] resubscribe[%p]", topic.c_str(), pDelegate));
            }
            else // 委托对象未订阅
            {
                modulelog_debug(STL_Text::format("topic[%s] subscribe[%p]", topic.c_str(), pDelegate));
                m_subscribeMap[topic].insert(pDelegate);
            }
            return 0; //已订阅主题，只需要登记委派对象
        }
        else // 主题未订阅
        {
            std::set<STL_MqttDelegate *> topicSet;
            topicSet.insert(pDelegate);
            modulelog_debug(STL_Text::format("topic[%s] new subscribe[%p]\n", topic.c_str(), pDelegate));
            m_subscribeMap[topic] = topicSet;

            //第二部：订阅
            {
                //第三步：检测连接
                if (m_status != STL_Mqtt::Status::Connected)
                {
                    fprintf(stderr, "mqtt subscribe <topic:%s>[%s](%s:%d) failed\n", topic.c_str(), m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort);
                    return -1;
                }
                else
                {
                    std::lock_guard<std::mutex> guard(m_mutex);
                    if (mosquitto_subscribe(m_pMqtt_st, NULL, topic.c_str(), qos) == MOSQ_ERR_SUCCESS)
                    {
                        modulelog_debug(STL_Text::format("mqtt subscribe <topic:%s>[%s](%s:%d) succeed\n", topic.c_str(), m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort));
                    }
                    else
                    {
                        modulelog_debug(STL_Text::format("mqtt subscribe <topic:%s>[%s](%s:%d) failed\n", topic.c_str(), m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort));
                    }
                }
            }
        }

        return 0;
    }

    int STL_Mqtt::unsubscribe(std::string topic, STL_MqttDelegate *pDelegate /* = nullptr8*/)
    {
        //第一步：检测连接
        if (m_status != STL_Mqtt::Status::Connected)
        {
            // fprintf(stderr, "[unsubscribe] mqtt isn`t Connented \n");
            return -1;
        }
        else
        {
            //第二步：检测订阅
            std::lock_guard<std::mutex> guard(m_subscribeMapMutex);
            if (m_subscribeMap.count(topic) == 1) // 主题已订阅
            {
                if (m_subscribeMap[topic].count(pDelegate) == 1) // 委托对象已订阅
                {
                    m_subscribeMap[topic].erase(pDelegate);
                }
                else // 委托对象未订阅
                {
                    modulelog_warn(STL_Text::format("pDelegate [%p] don't subscribe topic[%s], but to unsubscribe", pDelegate, topic));
                }
                if (m_subscribeMap[topic].empty())
                {
                    m_subscribeMap.erase(topic);
                    {
                        //第三步：取消订阅
                        std::lock_guard<std::mutex> guard(m_mutex);
                        mosquitto_unsubscribe(m_pMqtt_st, NULL, topic.c_str());
                    }
                }
                else // 还有委托对象订阅主题，不取消订阅
                {
                    //don`t to do anything
                }
            }
            else // 主题未订阅
            {
                modulelog_warn(STL_Text::format("don't subscribe topic[%s], but to unsubscribe[%p]", topic, pDelegate));
            }
            return 0;
        }
    }

    int STL_Mqtt::publishMessage(STL_Mqtt::Message message)
    {
        if (m_status != STL_Mqtt::Status::Connected)
        {
            // fprintf(stderr, "[publishMessage] mqtt isn`t Connented \n");
            return -1;
        }
        std::lock_guard<std::mutex> guard(m_mutex);
        mosquitto_publish(m_pMqtt_st, &(message.mid), message.topic.c_str(), message.payload.size(), message.payload.data(), message.qos, message.retain);

        STL_Mqtt::printMessage(message, "##publishMessage## ");
        return message.mid;
    }

    void STL_Mqtt::mqttThread()
    {
        void *UserData = (void *)this;

        // step1：初始化库,必须在调用库函数之前调用。 Must be called before any other mosquitto functions.
        STL_Mqtt::initlib();

        // step2: 创建客户端实例 Create a new mosquitto client instance.
        m_pMqtt_st = mosquitto_new(m_option.ClientID.c_str(), m_option.CleanSession, UserData);
        if (!m_pMqtt_st)
        {
            fprintf(stderr, "mosquitto_new: out of memory\n");
            return;
        }

        // int size = 65535;
        // mosquitto_opts_set(m_pMqtt_st, MOSQ_OPT_RECEIVE_MAXIMUM, (void *)(&size));
        // mosquitto_opts_set(m_pMqtt_st, MOSQ_OPT_SEND_MAXIMUM, (void *)(&size));
        // mosquitto_int_option(m_pMqtt_st, MOSQ_OPT_RECEIVE_MAXIMUM, 65535);
        // mosquitto_int_option(m_pMqtt_st, MOSQ_OPT_SEND_MAXIMUM, 65535);

        // step3: 设置回调
        mosquitto_user_data_set(m_pMqtt_st, UserData); //与mosquitto_new的第三个参数相同
        mosquitto_log_callback_set(m_pMqtt_st, STL_Mqtt::mqttLogCallback);
        mosquitto_connect_callback_set(m_pMqtt_st, STL_Mqtt::mqttConnectCallback);
        mosquitto_message_callback_set(m_pMqtt_st, STL_Mqtt::mqttMessageCallback);
        // mosquitto_subscribe_callback_set(m_pMqtt_st, STL_Mqtt::mqttSubscribeCallback);
        // mosquitto_unsubscribe_callback_set(m_pMqtt_st, STL_Mqtt::mqttUnsubscribeCallback);
        mosquitto_disconnect_callback_set(m_pMqtt_st, STL_Mqtt::mqttDisConnectCallback);

        // step4: 设置用户名密码
        if (mosquitto_username_pw_set(m_pMqtt_st, m_option.Username.c_str(), m_option.Password.c_str()) != MOSQ_ERR_SUCCESS)
        {
            fprintf(stderr, "set mqtt user & pwd error\n");
            return;
        }

        if (this->m_option.IsEnableTLS)
        {
            set_tls_connection(m_pMqtt_st, this->m_option.CAFilePath.c_str(), this->m_option.IsVerifyPeer);
        }

        setStatus(STL_Mqtt::Status::Connecting);
        // step4: 连接
        if (mosquitto_connect(m_pMqtt_st, m_option.BrokerAddress.c_str(), m_option.BrokerPort, m_option.KeepAliveInterval) != MOSQ_ERR_SUCCESS)
        {
            fprintf(stderr, "mqtt connect [%s](%s:%d) failed\n", m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort);
            setStatus(STL_Mqtt::Status::Over);
            return;
        }
        else
        {
            // setStatus(STL_Mqtt::Status::Connected);
            // modulelog_debug(STL_Text::format("mqtt connecting [%s](%s:%d) succeed\n", m_option.ClientID.c_str(), m_option.BrokerAddress.c_str(), m_option.BrokerPort));
        }

        // step5: 循环执行
        mosquitto_loop_forever(m_pMqtt_st, -1, 1);
        mosquitto_destroy(m_pMqtt_st);
        setStatus(STL_Mqtt::Status::Over);
    }

    void STL_Mqtt::modulelog_debug(std::string log)
    {
        if (m_logCallback)
        {
            m_logCallback(log);
        }
        else
        {
            stlog_debug("%s", log.c_str());
        }
    }

    void STL_Mqtt::modulelog_warn(std::string log)
    {
        if (m_logCallback)
        {
            m_logCallback(log);
        }
        else
        {
            stlog_warn("%s", log.c_str());
        }
    }

    void STL_Mqtt::mqttLogCallback(struct mosquitto *mosq, void *userdata, int level, const char *str)
    {
        STL_Mqtt *self = (STL_Mqtt *)userdata;
        self->modulelog_debug(STL_Text::format("[tag:%d][level:%d]%s\n", self->m_status, level, str));
    }

    void STL_Mqtt::printMessage(STL_Mqtt::Message message, std::string tag /*= "Message"*/)
    {
        modulelog_debug(STL_Text::format("%s[mid:%d][topic:(%s)][qos:%d][retain:%d]\n%s\n",
                                         tag.c_str(),
                                         message.mid,
                                         message.topic.c_str(),
                                         message.qos,
                                         message.retain,
                                         message.payload.c_str()));
    }

    std::string STL_Mqtt::getMessageLog(STL_Mqtt::Message message, std::string tag /*= "Message"*/)
    {
        char ret[1024] = {0};
        snprintf(ret, 1023, "%s[mid:%d][topic:(%s)][qos:%d][retain:%d]\n%s\n",
                 tag.c_str(),
                 message.mid,
                 message.topic.c_str(),
                 message.qos,
                 message.retain,
                 message.payload.c_str());
        return std::string(ret);
    }

    bool MqttTopicMatch(std::string subscribeTopic, std::string messageTopic)
    {
        // 只支持尾部#号通配符
        auto pos = subscribeTopic.find("#");
        if (subscribeTopic.find("#") == std ::string::npos)
        {
            return (subscribeTopic == messageTopic);
        }
        else
        {
            auto temp = subscribeTopic.substr(0, pos);
            return STL_Text::string_start_with(messageTopic, temp);
        }
    }

    void STL_Mqtt::OnMqttMessage(STL::STL_Mqtt::Message message)
    {
        { // 控制守卫域
            std::lock_guard<std::mutex> guard(m_subscribeMapMutex);
            for (auto itTopic = m_subscribeMap.begin(); itTopic != m_subscribeMap.end(); ++itTopic)
            {
                if (MqttTopicMatch(itTopic->first, message.topic))
                {
                    for (std::set<STL_MqttDelegate *>::iterator it = m_subscribeMap[itTopic->first].begin(); it != m_subscribeMap[itTopic->first].end(); ++it)
                    {
                        STL_MqttDelegate *target = (*it);
                        if (target)
                        {
#if STL_SYNC_MESSAGE_HANDLE // 同步处理
                            target->OnMqttMessage(message);
#else //异步处理
                            std::string name = std::string("[") + message.topic + std::string("]") + message.payload;
                            STL_ASYNC_TIMEOUT(name, 10, STL_Mqtt::AsyncMqttMessageCallback, target, message);
#endif
                        }
                    }
                }
            }
        }
    }

    void STL_Mqtt::AsyncMqttMessageCallback(STL_MqttDelegate *target, STL::STL_Mqtt::Message message)
    {
        target->OnMqttMessage(message);
    }

    void STL_Mqtt::mqttMessageCallback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
    {
        STL_Mqtt *self = (STL_Mqtt *)userdata;
        if (message->payloadlen)
        {
            STL::STL_Mqtt::Message ret_message;
            ret_message.topic = std::string(message->topic);
            ret_message.payload = std::string((char *)message->payload, message->payloadlen);
            ret_message.mid = message->mid;
            ret_message.qos = message->qos;
            ret_message.retain = message->retain;
            self->modulelog_debug(STL_Text::format("[mid:%d][topic:(%s)] payload :%s\n", ret_message.mid, ret_message.topic.c_str(), ret_message.payload.c_str()));
            self->OnMqttMessage(ret_message);
        }
        else
        {
            self->modulelog_warn(STL_Text::format("[mid:%d][topic:(%s)] payload empty\n", message->mid, message->topic));
        }
    }

    void STL_Mqtt::mqttConnectCallback(struct mosquitto *mosq, void *userdata, int result)
    {
        STL_Mqtt *self = (STL_Mqtt *)userdata;
        self->modulelog_debug(STL_Text::format("mqtt connect callback result = %d\n", result));
        if (!result)
        {
            self->modulelog_debug(STL_Text::format("[callback] mqtt connect [%s](%s:%d) succeed\n", self->m_option.ClientID.c_str(), self->m_option.BrokerAddress.c_str(), self->m_option.BrokerPort));
            self->setStatus(STL_Mqtt::Status::Connected);
            self->initSubscribe();
            //TODO 连接回调
        }
        else
        {
            self->modulelog_debug(STL_Text::format("[callback] mqtt connect [%s](%s:%d) failed\n", self->m_option.ClientID.c_str(), self->m_option.BrokerAddress.c_str(), self->m_option.BrokerPort));
        }
    }

    void STL_Mqtt::mqttDisConnectCallback(struct mosquitto *mosq, void *userdata, int result)
    {
        STL_Mqtt *self = (STL_Mqtt *)userdata;
        self->modulelog_warn(STL_Text::format("mqtt disconnect callback result = %d\n", result));

        if (!result)
        {
            self->setStatus(STL_Mqtt::Status::Disconnected);
            //TODO 断开连接回调
        }
        else
        {
            //TODO 失败处理
            self->setStatus(STL_Mqtt::Status::Over);
        }
    }

    void STL_Mqtt::mqttSubscribeCallback(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)
    {
        // STL_Mqtt *self = (STL_Mqtt *)userdata;

        // modulelog_debug(STL_Text::format("Subscribed Callback [mid: %d][qos_count:%d]: ", mid, qos_count));
        // for (int i = 0; i < qos_count; i++)
        // {
        //     modulelog_debug(STL_Text::format(", %d", granted_qos[i]));
        // }
        // modulelog_debug(STL_Text::format("\n"));
    }

    void mqttUnsubscribeCallback(struct mosquitto *mosq, void *userdata, int mid)
    {
        // STL_Mqtt *self = (STL_Mqtt *)userdata;
    }
} // namespace STL